package com;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.concurrent.TimeUnit;

public class FlinkJsonToCsvStreamingJob {
    private static final Logger LOG = LoggerFactory.getLogger(FlinkJsonToCsvStreamingJob.class);

    public static void main(String[] args) throws Exception {
        // 创建 Flink 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 通过自定义 SourceFunction 创建一个无限数据流
        DataStream<String> jsonStream = env.addSource(new JsonSourceFunction());

        // 处理 JSON 数据
        DataStream<String> csvStream = jsonStream.map(new JsonToCsvMapper());

        // 触发计算并打印
        csvStream.print(); // 本地调试使用
        env.execute("Flink JSON to CSV Streaming Job");
    }

    /**
     * 自定义 Flink SourceFunction, 每 3 秒生成一次 JSON 数据
     */
    public static class JsonSourceFunction implements SourceFunction<String> {
        private volatile boolean running = true;

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            while (running) {
                // 模拟 JSON 数据
                String jsonInput = "{\"id\":3,\"productId\":\"N19639\",\"productName\":\"镇海农商银行招宝A+2020047期人民币封闭净值型理财产品\",\"bank\":\"镇海农商行\",\"currency\":\"人民币\",\"issueDate\":\"2020-09-15\"}";

                // 发送数据
                ctx.collect(jsonInput);

                // 等待 3 秒
                TimeUnit.SECONDS.sleep(3);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

    /**
     * 处理 JSON 并转换为 CSV 格式
     */
    public static class JsonToCsvMapper implements MapFunction<String, String> {
        private static final ObjectMapper objectMapper = new ObjectMapper();

        @Override
        public String map(String value) throws Exception {
            Map<String, Object> jsonMap = objectMapper.readValue(value, Map.class);

            // 拼接 value 为 CSV 格式
            String csvOutput = jsonMap.values().stream()
                    .map(Object::toString)
                    .collect(Collectors.joining(","));

            // 打印日志
            LOG.info("Processed CSV Output: {}", csvOutput);

            return csvOutput;
        }
    }
}
